package com.ruoyi.learn.java.concurrent;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;

import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

import static java.lang.Thread.sleep;

/**
 * 题目：模拟股票交易系统，实现订单匹配引擎，处理买卖订单的撮合。
 * 视频接入
 */
@Data
@Slf4j
public class OrderMatchingEngine {

    public static void main(String[] args) {
        //
        LinkedBlockingQueue<CallFrom> callFromLinkedBlockingQueue = new LinkedBlockingQueue<CallFrom>();

        LinkedBlockingQueue<Receiver> receiverLinkedBlockingQueue = new LinkedBlockingQueue<Receiver>();

        ExecutorService executorService = Executors.newFixedThreadPool(10);
        // 呼叫者模拟
        for (int i = 0; i < 4; i++) {
            executorService.execute(getCallFromLinkedBlockingQueueRunnable(callFromLinkedBlockingQueue));
        }

        // 接听者模拟
        for (int i = 0; i < 2; i++) {
            executorService.execute(getReceiverLinkedBlockingQueueCommand(receiverLinkedBlockingQueue));
        }

        // 接听者处理逻辑
        for (int i = 0; i < 2; i++) {
            executorService.execute(toCall(callFromLinkedBlockingQueue, receiverLinkedBlockingQueue));
        }


    }

    /**
     * 呼叫逻辑实现
     * @param callFromLinkedBlockingQueue
     * @param receiverLinkedBlockingQueue
     * @return
     */
    private static Runnable toCall(LinkedBlockingQueue<CallFrom> callFromLinkedBlockingQueue, LinkedBlockingQueue<Receiver> receiverLinkedBlockingQueue) {
        return () ->
        {
            while (true) {
                // 有呼叫者并且有接听者
                if (!callFromLinkedBlockingQueue.isEmpty() && !receiverLinkedBlockingQueue.isEmpty()) {
                    // 安排进行视频
                    // 呼叫者跟接听者2个队列
                    // 撮合机制，批量随机撮合？
                    // 异常情况怎么处理?从队列拿了但是线程挂掉了？  持久化？
                    // 消息队列？
                    // 队列在mysql也有一份？，有变动才会更新mysql，如果发现入队的优先级高的呼叫者跟坐席迟迟没执行完就属于异常数据？
                    // 系统重启咋办？数据丢失？  补偿机制，如果mysql里面存在优先级高的呼叫者跟坐席迟迟没配对成功就重新检查是否在队列里面，如果不在队列里面这个时候就重先入队（优先级低的反而已经处理了？或者
                    // 就一个人得情况下  redis zset队列里没这个人,但mysql这个人的数据还在持续多少秒可以让他重先入队
                    // ）


                    // poll非阻塞移除
                    // peel 不移除
                    // take 阻塞移除
                    CallFrom callFrom = null;
                    try {
                        callFrom = callFromLinkedBlockingQueue.take();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    Receiver receiver = null;
                    try {
                        receiver = receiverLinkedBlockingQueue.take();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }

                    if (receiver != null && callFrom != null) {
//                                log.info("配对成功，receiver={}，callFrom={}", JSON.toJSONString(receiver),JSON.toJSONString(callFrom));
                        log.info("配对成功，receiver={}，callFrom={}", receiver, callFrom);
                        // 加入正在配对队列
                        // 执行配对，配对成功后，配对的过程一直在维护一个刷新时间，若这个时间超过10秒未更新，可以考虑刷新失败，启动补偿机制
                        // 有个合理的超时时间？30秒还没完成可以认为是超时，若超时则继续放在队列里面优先处理
                        // 此时需要更新mysql的状态，本次通话配对完成

                        log.info("callFromLinkedBlockingQueue={},receiverLinkedBlockingQueue={}", callFromLinkedBlockingQueue.size(), receiverLinkedBlockingQueue.size());
                    } else {
                        // 配对失败，将未配对的塞入队列继续匹配
                        log.info("配对失败、单线程状态下不会发生，receiver={}，callFrom={}", receiver, callFrom);
                    }
                } else {
                    // 呼叫者来了
                    try {
                        sleep(1000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            }
        };
    }

    private static Runnable getReceiverLinkedBlockingQueueCommand(LinkedBlockingQueue<Receiver> receiverLinkedBlockingQueue) {
        return () -> {

            while (true) {
                try {
                    sleep(RandomUtils.nextLong(0, 1000));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                // 接听者来了
                if (receiverLinkedBlockingQueue.size() <= 100) {
                    Receiver receiver = new Receiver();
                    receiverLinkedBlockingQueue.add(receiver);
                    log.info("接听者来了{},receiverLinkedBlockingQueue={}", receiver, receiverLinkedBlockingQueue.size());
                }
            }
        };
    }

    private static Runnable getCallFromLinkedBlockingQueueRunnable(LinkedBlockingQueue<CallFrom> callFromLinkedBlockingQueue) {
        return () ->
        {
            while (true) {
                try {
                    sleep(RandomUtils.nextLong(0, 100));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                // 呼叫者来了
                if (callFromLinkedBlockingQueue.size() <= 10000000) {
                    CallFrom callFrom = new CallFrom();
                    callFromLinkedBlockingQueue.add(callFrom);
                    log.info("呼叫者来了{},callFromLinkedBlockingQueue={}", callFrom, callFromLinkedBlockingQueue.size());
                }
            }
        };
    }

    @Builder
    @ToString
    @NoArgsConstructor
    @AllArgsConstructor
    @Data
    public static  class CallFrom{
        /**
         * 用户名
         */
        private String userName = UUID.randomUUID().toString();

        /**
         * 用户ID
         */
        private Long userId = RandomUtils.nextLong();

        /**
         * 会议编号
         */
        private String meetingNo= UUID.randomUUID().toString();

        /**
         * 呼叫时间
         */
        private LocalDateTime callTime = LocalDateTime.now();



        /**
         * 接受时间
         */
        private LocalDateTime receiveTime = LocalDateTime.now();

        public String toJsonString() {
            try {
                return new ObjectMapper().writeValueAsString(this);
            } catch (Exception e) {
                return toString(); // 回退到普通toString
            }
        }
    }

    /**
     * 坐席接听逻辑
     * ZRANGE players 0 0  待接听坐席信息全部放在ZSET里面，每次有视频接入时利用
     * ZPOPMIN  players 0 0  取分数最低的并从队列里弹出跟该人员进行视频，若对接成功，则更新mysql的状态
     * 若对接失败，则小数位数的随机数重先计算，放入ZSET队列，此时也可更新mysql  失败次数+1
     *
     */
    @Builder
    @ToString
    @NoArgsConstructor
    @AllArgsConstructor
    @Data
    public static class Receiver{

        /**
         * 等待呼叫的坐席方放在redis Zset里面，保证分布式
         * mysql维护状态方便查看，有新的呼入或者某一段呼叫结束会去更新mysql里面数据的状态
         * 时长（秒）+随机数（为了保证公平，每次进入队列都有个小数的随机数）
         */
        private Double score = RandomUtils.nextDouble();

        /**
         * 用户名
         */
        private String userName = UUID.randomUUID().toString();

        /**
         * 用户ID
         */
        private Long userId = RandomUtils.nextLong();

        /**
         * 会议编号
         */
        private String meetingNo= UUID.randomUUID().toString();

        /**
         * 呼叫时间
         */
        private LocalDateTime callTime = LocalDateTime.now();



        /**
         * 接受时间
         */
        private LocalDateTime receiveTime = LocalDateTime.now();

        public String toJsonString() {
            try {
                return new ObjectMapper().writeValueAsString(this);
            } catch (Exception e) {
                return toString(); // 回退到普通toString
            }
        }
    }

}
